package utils

import (
	"sync"
	"time"
)

type Consumer interface {
	Execute([][]byte)
	Done()
}

type WaitQueue struct {
	queue    *BlockQueue
	lock     *sync.RWMutex
	cond     *sync.Cond
	seq      uint32
	close    bool
	consumer Consumer
	inner    uint32
	pre      time.Time
	minSync  int64
}

func NewWaitQueue(size int, consumer Consumer, minPoll time.Duration, minSync time.Duration) *WaitQueue {
	lock := new(sync.RWMutex)
	q := &WaitQueue{
		queue:    NewQueue(size, minPoll),
		lock:     lock,
		cond:     sync.NewCond(lock.RLocker()),
		seq:      1,
		close:    false,
		consumer: consumer,
		inner:    1,
		pre:      time.Now(),
		minSync:  int64(minSync),
	}
	// bigger than 1us
	if minSync >= time.Microsecond {
		afterPublish(q, minSync)
	} else {
		minSync = 0
	}
	return q
}

func afterPublish(q *WaitQueue, duration time.Duration) {
	d := int64(duration)
	go func() {
		for {
			delta := time.Now().Sub(q.pre)
			if delta.Nanoseconds() < d {
				time.Sleep(duration - delta)
			} else {
				q.publishInner()
			}
		}
	}()
}

// SaveOrPublish 保存seq的最大值或者发布seq
// 当小于某个时间阈值时, 将seq统一进行保存, 否则直接发布seq
// 这样做的原因是: 与磁盘交互时,避免过于频繁的sync
func (q *WaitQueue) saveOrPublish(seq uint32) {
	delta := time.Now().Sub(q.pre)
	if seq > q.inner {
		q.inner = seq
	}
	if delta.Nanoseconds() < q.minSync {
		return
	} else {
		q.publishInner()
	}
}

func (q *WaitQueue) publishInner() {
	if q.inner > q.seq {
		q.consumer.Done()
	}
	q.publish(q.inner)
	q.pre = time.Now()
}

func (q *WaitQueue) Start() {
	go func() {
		for true {
			if q.close {
				break
			}
			batch, seq := q.queue.PollAll()
			q.consumer.Execute(batch)
			q.saveOrPublish(seq)
		}
	}()
}

func (q *WaitQueue) publish(seq uint32) {
	q.lock.Lock()
	q.seq = seq
	q.lock.Unlock()
	q.cond.Broadcast()
}

func (q *WaitQueue) Add(elem []byte) uint32 {
	return q.queue.Add(elem)
}

func (q *WaitQueue) WaitDone(seq uint32) {
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	for seq >= q.seq {
		q.cond.Wait()
	}
}
